package util;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.SocketException;
import java.nio.channels.ClosedByInterruptException;
import java.util.Collection;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

import javax.xml.bind.JAXBException;

/**
 * A handler that reads over registered devices and offloads processing to an abstract method.
 * <br>
 * It is up to the user that implements a subclass how to perform write operations.
 * To this end, the {@link #getInterface(Integer)} method is provided.
 * <br>
 * A thread is created to handle reading from each device. Every such thread is created when {@link #run()} is called and closed when {@link #run()} finishes.
 * The thread calling {@link #run()} will also be the thread processing messages, and will continue until another thread calls {@link #stop()}.
 * @author Bitidork
 *
 */
public abstract class MessageHandler {
	/**
	 * Initializes a handler with an empty set of registered devices and an empty queue of unprocessed messages.
	 */
	public MessageHandler( ) {
		this.devices = new ConcurrentHashMap<Integer, MBMessageChannelInterface>( );
		this.unprocessedMessages = new LinkedBlockingQueue<MessageEntry>( );
		this.readingThreads = new LinkedList<Thread>( );
		this.isRunning = false;
		this.isStopped = true;
	}
	
	/**
	 * Registers the device with the given message interface.
	 * @param device The device to register.
	 * @param msgInterface The {@link MBMessageChannelInterface} associated with this device.
	 * @throws IllegalArgumentException if the device or interface is already registered.
	 */
	public final void registerDevice( final int device, final MBMessageChannelInterface msgInterface ) {
		if ( this.devices.containsKey( device ) )
			throw new IllegalArgumentException("Device (" + device +") is already registered with this handler.");
		else if ( this.devices.containsValue( msgInterface ) )
			throw new IllegalArgumentException("The supplied message interface is already registered with this handler.");
		
		this.devices.put( device, msgInterface );
	}
	
	/**
	 * Reads a message from a device and enqueues it for processing.
	 * @param dev The device to read from.
	 * @throws InterruptedIOException if the read was interrupted.
	 * @throws JAXBException if an error was encountered while unmarshalling a message.
	 * @throws IOException if an error was encountered while reading data from the channel.
	 */
	private final void readDevice( final Integer dev ) throws InterruptedIOException, InterruptedException, JAXBException, IOException {
		MBMessageChannelInterface msgInterface = this.getInterface( dev );
		
		// stop if device is not registered
		if ( msgInterface == null )
			return;
		
		// read a message
		MBMessage msg = msgInterface.read( );
		
		// enqueue message
		this.unprocessedMessages.add( new MessageEntry( dev, msg ) );
	}
	
	/**
	 * Sets the current thread to create channel-readers and start processing.
	 * @throws InterruptedException if the thread was interrupted while waiting for data on a queue.
	 */
	public void run( ) throws InterruptedException {
		this.setupReadingThreads( );
		
		// set running states 
		synchronized (this) {
			this.isRunning = true;
			this.isStopped = false;
		}
		
		try {
			this.startReadingThreads( );
			
			// while running, process messages
			while ( this.isRunning() ) {
				final MessageEntry me = this.unprocessedMessages.take( );
				
				this.processMessage( me.device, me.msg );
			}
		} finally {
			this.stopReadingThreads( );
			
			// stopped running, set flag and alert all threads waiting for this object
			synchronized (this) {
				this.isRunning = false;
				this.isStopped = true;
				this.notifyAll( );
			}
		}
	}
	
	/**
	 * Sets up the threads for reading devices.
	 * Each thread that is created is added to {@link #readingThreads}.
	 */
	private void setupReadingThreads( ) {
		lib.debug( "Setting up threads for reading devices [" + this.readingThreads.size() + "]", lib.FLAG_DEBUG_MSG_HANDLER);
		for ( Integer dev : this.devices.keySet( ) ) {
			final MBMessageChannelInterface msgInterface = this.devices.get( dev );
			if ( msgInterface == null )
				continue;
			
			final Thread thread = new Thread( new ChannelReader( this, dev ) );
			readingThreads.add( thread );
		}
		lib.debug( "...Finished setting up reading threads", lib.FLAG_DEBUG_MSG_HANDLER );
	}
	
	/**
	 * Starts each thread set up in {@link #setupReadingThreads()}.
	 */
	private void startReadingThreads( ) {
		lib.debug( "Starting threads for reading devices [" + this.readingThreads.size() + "]", lib.FLAG_DEBUG_MSG_HANDLER );
		for ( Thread thread : this.readingThreads ) {
			thread.run( );
		}
		lib.debug( "...Finished starting reading threads", lib.FLAG_DEBUG_MSG_HANDLER );
	}
	
	/**
	 * Stops each thread set up in {@link #setupReadingThreads()} and started in {@link #startReadingThreads()}.
	 * Blocks until this operation is complete.
	 */
	private void stopReadingThreads( ) {
		lib.debug( "Stopping reading threads", lib.FLAG_DEBUG_MSG_HANDLER );
		// give each thread the interrupt
		for ( Thread thread : this.readingThreads ) {
			// interrupt thread
			thread.interrupt( );
		}
		
		// wait for threads to die
		for ( Thread thread : this.readingThreads ) {
			while ( thread.isAlive( ) ) {
				try {
					thread.join( );
				} catch ( InterruptedException e ) {
					// wait for  close
					lib.debug( "Attempted to stop thread but was interrupted; retrying.", lib.FLAG_DEBUG_RETRY_ERRORS );
				}
			}
		}
		
		this.readingThreads.clear( );
		lib.debug( "Finished stopping reading threads", lib.FLAG_DEBUG_MSG_HANDLER );
	}
	
	/**
	 * A mapping between a device and its channel interface.
	 * This mapping is a bijection of sets.
	 * This mapping is thread-safe.
	 */
	private Map<Integer, MBMessageChannelInterface> devices;
	
	/**
	 * A queue of messages to process.
	 * This queue is thread-safe.
	 */
	private BlockingQueue<MessageEntry> unprocessedMessages;
	
	/**
	 * A collection of threads that are setup to read.
	 * Before calling {@link Thread#join()}, check {@link Thread#isAlive()}.
	 * This collection is not thread-safe.
	 */
	private Collection<Thread> readingThreads;
	
	/**
	 * Determines whether or not this handler should be processing messages.
	 * In order to provide concurrency, do not access this field directly.
	 */
	private boolean isRunning;
	
	/**
	 * Determines whether or not this handler has stopped processing messages.
	 * This is set when the handler has perceived the change in state of {@link #isRunning}.
	 * In order to provide concurrency, do not access this field directly.
	 */
	private boolean isStopped;
	
	/**
	 * Gets the message channel associated with the device.
	 * @param dev The device.
	 * @return The message channel associated with the device, or null if the device is not registered.
	 */
	MBMessageChannelInterface getInterface( final Integer dev ) {
		return this.devices.get( dev );
	}
	
	/**
	 * Flags the handler to stop running.
	 * This method blocks until the thread that has called {@link #run()} has returned.
	 * @throws InterruptedException if the thread was interrupted while waiting for the thread to return.
	 */
	public synchronized void stop( ) throws InterruptedException {
		if ( this.isStopped )
			return;
		
		this.isRunning = false;
		this.wait( );
	}
	
	/**
	 * Gets whether or not the handler should be processing messages.
	 * @return Returns true if and only if the handler should be processing messages.
	 */
	private synchronized boolean isRunning( ) {
		return this.isRunning;
	}
	
	/**
	 * Gets whether or not the handler is processing messages.
	 * @return Returns false if and only if the handler is processing messages.
	 */
	public synchronized boolean isStopped( ) {
		return this.isStopped;
	}
	
	/**
	 * A device-message pair, where the device is the immediate source of the message.
	 * @author Bitidork
	 *
	 */
	private final class MessageEntry {
		/**
		 * The device that the message came from.
		 */
		public Integer device;
		
		/**
		 * The message.
		 */
		public MBMessage msg;
		
		/**
		 * Constructs a message entry.
		 * @param device The entry's device.
		 * @param msg The entry's message.
		 */
		public MessageEntry( final Integer device, final MBMessage msg ) {
			this.device = device;
			this.msg = msg;
		}
	}
	
	/**
	 * A runnable set up to read a device on a {@link MessageHandler}.
	 * If the reader encounters an exception, it will close the channel and quit.
	 * The reader will report expected exceptions to {@link MessageHandler#onParsingError(Integer, JAXBException)}
	 * , {@link MessageHandler#onTimeoutError(Integer, InterruptedIOException)}, {@link MessageHandler#OnUnknownIOError(Integer, IOException)}.
	 * @author Bitidork
	 *
	 */
	public final class ChannelReader implements Runnable {
		/**
		 * The handler the reader is attached to.
		 */
		private MessageHandler handler;
		
		/**
		 * The device the reader is attached to.
		 */
		private Integer device;
		
		/**
		 * The message interface of the device the reader is attached to.
		 */
		private MBMessageChannelInterface msgInterface;
		
		/**
		 * Constructs a channel reader for the given device on the handler.
		 * @param handler The handler.
		 * @param device The device.
		 */
		public ChannelReader( final MessageHandler handler, final Integer device ) {
			if ( handler == null )
				throw new IllegalArgumentException( "The supplied handler is null." );
			
			this.handler = handler;
			this.device = device;
			this.msgInterface = handler.getInterface( device );
			
			if ( msgInterface == null )
				throw new IllegalArgumentException( "The device (" + device + ") is not registered with the handler.");
		}
		
		@Override
		public void run( ) {
			try {
				lib.debug("Reading thread started on device (" + device +")", lib.FLAG_DEBUG_MSG_HANDLER);
				handler.onReaderStart( device, this );
				
				// used by the error handlers to determine if the thread should close
				boolean run = true;
				
				// read messages and enqueue them
				while ( handler.isRunning( ) && run ) {
					// handle possibly non-fatal exceptions
					try {
						handler.readDevice( device );
					} catch ( JAXBException e ) {
						// report exception to handler, parsing error
						run = handler.onParsingError( device, e );
					} catch ( InterruptedIOException e ) {
						// report timeout exception; possible disconnect
						run = handler.onTimeoutError( device, e );
					}
				}
			} catch ( SocketException e ) {
				// socket was closed (locally) while reading
				lib.debug("Socket closed on device (" + device +")", lib.FLAG_DEBUG_MSG_HANDLER);
			} catch ( ClosedByInterruptException e ) {
				// quit and clean up; forced quit
				lib.debug("Forced quit on device (" + device +")", lib.FLAG_DEBUG_MSG_HANDLER);
			} catch ( InterruptedException e ) {
				// quit and clean up; forced quit
				lib.debug("Forced quit on device (" + device +")", lib.FLAG_DEBUG_MSG_HANDLER);
			}  catch ( IOException e ) {
				// report exception; broken channel
				lib.debug("Unknown IO error on device (" + device +"): " + e.toString( ) , lib.FLAG_DEBUG_MSG_HANDLER);
				handler.OnUnknownIOError( device, e );
			} finally {
				lib.debug("Reading thread closing on device (" + device +")", lib.FLAG_DEBUG_MSG_HANDLER);
				handler.onReaderClose( device, this );
				msgInterface.close( ); // may be redundant, but for consistency
			}
		}
		
	}
	
	/**
	 * Called when a thread that is reading the device is starting.
	 * This method is called by the reading thread and should be thread-safe.
	 * @param dev The device.
	 * @param reader The channel reader for the device.
	 */
	public abstract void onReaderStart( final Integer dev, final ChannelReader reader );
	
	/**
	 * Called when a thread that is reading the device is closing.
	 * This method is called by the reading thread and should be thread-safe.
	 * @param dev The device.
	 * @param reader The channel reader for the device.
	 */
	public abstract void onReaderClose( final Integer dev, final ChannelReader reader );
	
	/**
	 * Called whenever a parsing error has occurred in the reader for the device.
	 * This method is called by the reading thread and should be thread-safe.
	 * @param dev The device.
	 * @param e The exception thrown.
	 * @return True if the reader should ignore the error.
	 */
	public abstract boolean onParsingError( final Integer dev, final JAXBException e );
	
	/**
	 * Called whenever a timeout error has occurred in the reader for the device.
	 * This method is called by the reading thread and should be thread-safe.
	 * @param dev The device.
	 * @param e The exception thrown.
	 * @return True if the reader should ignore the error.
	 */
	public abstract boolean onTimeoutError( final Integer dev, final InterruptedIOException e );
	
	/**
	 * Called whenever any other IO error has occurred in the reader for the device.
	 * This method is called by the reading thread and should be thread-safe.
	 * @param dev The device.
	 * @param e The exception thrown.
	 */
	public abstract void OnUnknownIOError( final Integer dev, final IOException e );
	
	/**
	 * This method is called when a message has arrived from the device and needs to be processed.
	 * @param msg The message.
	 * @param device The device the messaged was received from.
	 */
	public abstract void processMessage( final Integer dev, final MBMessage msg );
}
